package myapps3;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class KafkaIntecepter implements ProducerInterceptor {
	private volatile long sendsuccess = 0;
	private volatile long sendfail = 0;
	@Override
	public ProducerRecord onSend(ProducerRecord record) {
		Object value = "pre_"+record.value();
		return new ProducerRecord(record.topic(),record.partition(),record.timestamp(),record.key(),value,record.headers());
	}

	@Override
	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
		if (exception == null){
			sendsuccess ++;
		}else{
			sendfail ++;
		}

	}

	@Override
	public void close() {
		double ratio = sendsuccess/(sendsuccess+sendfail);
		System.out.println("成功率："+ratio);
	}

	@Override
	public void configure(Map<String, ?> configs) {

	}
}
